package cn.doitedu.api;

import beans.UserAction;
import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/2/28
 * @Desc: 学大数据，上多易教育
 *   测流输出演示
 **/
public class _20_SideOutput_Demo {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);


        // json
        // a,1
        DataStreamSource<String> stream = env.socketTextStream("localhost", 8899);

        SingleOutputStreamOperator<UserAction> mainStream = stream.process(new ProcessFunction<String, UserAction>() {
            @Override
            public void processElement(String element, ProcessFunction<String, UserAction>.Context ctx, Collector<UserAction> collector) throws Exception {

                try {
                    UserAction userAction = JSON.parseObject(element, UserAction.class);
                    // 利用 collector 输出的就在主流中
                    collector.collect(userAction);

                } catch (Exception e) {

                    // 利用ctx.output 输出的数据 在一个侧流中，而且有一个标签  no_json
                    if (element.startsWith("a")) {
                        ctx.output(new OutputTag<String>("a_no_json") {}, element);
                    } else {
                        ctx.output(new OutputTag<String>("b_no_json") {}, element);
                    }
                }
            }
        });


        // 获取侧流数据
        // OutputTag 要new成一个实现
        SideOutputDataStream<String> aNoJson = mainStream.getSideOutput(new OutputTag<String>("a_no_json") {
        });
        SideOutputDataStream<String> bNoJson = mainStream.getSideOutput(new OutputTag<String>("b_no_json") {
        });

        SingleOutputStreamOperator<String> s1 = aNoJson.map(s -> s);
        SingleOutputStreamOperator<String> s2 = bNoJson.map(s -> s);

        mainStream.print("main==>");

        aNoJson.print("a_side==>");

        bNoJson.print("b_side==>");


        env.execute();
    }

}
